package com.urfresh.sp.flume.receive.stream;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;

import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;

/**
 * Created by urfresh.mark on 2017/5/11.
 */
public class KafkaStream {

    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "120.26.223.176:9092");
//        props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "120.26.223.176:2181");
        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
//        props.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class);


        // setting offset reset to earliest so that we can re-run the demo code with the same pre-loaded data
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        KStreamBuilder builder = new KStreamBuilder();

        final Serde<String> stringSerde = Serdes.String();
        final Serde<Long> longSerde = Serdes.Long();
        KStream<String, String> source = builder.stream(stringSerde, stringSerde, "urfresh_user_info");

        KTable<String, Long> wordCounts = source
                .flatMapValues(new ValueMapper<String, Iterable<String>>() {
                    @Override
                    public Iterable<String> apply(String value) {
                        System.out.println("apply="+value);
                        return Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" "));
                    }
                }).map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(String key, String value) {
                        System.out.println("key="+key+",value="+value);
                        return new KeyValue<>(value, value);
                    }
                })
//                .countByKey("click");
                .groupByKey()
                .count("click");


//        KStream<String, String> textLines = builder.stream(stringSerde, stringSerde, "urfresh_user_info");
//
//        KTable<String, Long> wordCounts = textLines
//                // Split each text line, by whitespace, into words.
//                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
//
//                // Group the text words as message keys
//                .groupBy((key, value) -> value)
//
//                // Count the occurrences of each word (message key).
//                .count("click");


        System.out.println(wordCounts);
//        wordCounts.to(stringSerde, longSerde, "streams-wordcount-processor-output");


        // need to override value serde to Long type
        wordCounts.to(stringSerde, longSerde, "streams-wordcount-processor-output");

        KafkaStreams streams = new KafkaStreams(builder, props);
        streams.start();

        // usually the stream application would be running forever,
        // in this example we just let it run for some time and stop since the input data is finite.
//        Thread.sleep(1000L);

//        streams.close();
    }

}
